/*
 * chat_epoll.c
 *
 *  Created on: 2012-6-26
 *      Author: rendong
 */

#include "chat_epoll.h"
#include "../os/chat_file.h"

static ssize_t write_n(chat_server_config_t *config, int fd, const void *buffer, size_t n);

/**
 * 启动socket服务
 */
void chat_epoll_socket_start(chat_server_config_t *config)
{
	int listenfd, readyfds, i, connfd, sockfd,nreadbytes=0,flag=1,len=sizeof(int), buffersize, old_buffer_data_size;
	int writebytes, n, tn;

	struct epoll_event ev, events[EPOLL_EVENT_SIZE];

	struct sockaddr_in clientaddr;
	struct sockaddr_in serveraddr;

	time_t now;
	struct tm *now_time;

	int keepalive = 1; // 开启keepalive属性
	int keepidle = 300; // 如该连接在300秒内没有任何数据往来,则进行探测
	int keepinterval = 5; // 探测时发包的时间间隔为5 秒
	int keepcount = 3; // 探测尝试的次数.如果第1次探测包就收到响应了,则后2次的不再发.


	socklen_t clilen = sizeof(struct sockaddr_in);

	chat_session_t * session;

	if (config->handler_new == NULL || config->handler_read == NULL || config->handler_close == NULL){
		chat_log_error("config error hander is NULL\n");
		exit(1);
	}

	config->srv_epfd = epoll_create(EPOLL_EVENT_SIZE);

	listenfd = socket(AF_INET, SOCK_STREAM, 0);

	if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &flag, len) == -1){
		chat_log_error("set sockopt SO_REUSEADDR error\n");
	}

	setnonblocking(listenfd);

	// 设置超时处理
	setsockopt(listenfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive , sizeof(keepalive ));
	setsockopt(listenfd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepidle , sizeof(keepidle ));
	setsockopt(listenfd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepinterval , sizeof(keepinterval ));
	setsockopt(listenfd, SOL_TCP, TCP_KEEPCNT, (void *)&keepcount , sizeof(keepcount ));

	memset(&ev, 0, sizeof(struct epoll_event));

	ev.data.fd = listenfd;
	ev.events=EPOLLIN;

	// 监听端口
	epoll_ctl(config->srv_epfd,EPOLL_CTL_ADD,listenfd,&ev);

	bzero(&serveraddr, sizeof(serveraddr));
	serveraddr.sin_family = AF_INET;
	inet_aton(config->srv_addr,&(serveraddr.sin_addr));
	serveraddr.sin_port=htons(config->srv_port);

	if (bind(listenfd,(struct sockaddr *)&serveraddr, sizeof(serveraddr))){
		chat_log_error("bind fail %s:%d errno:%d err:%s\n", config->srv_addr, config->srv_port, errno, strerror(errno));
		exit(1);
	}

	chat_log_debug("start listen %s:%d backlog:%d\n", config->srv_addr, config->srv_port, LISTEN_BACKLOG);

	if (listen(listenfd, LISTEN_BACKLOG)){
		chat_log_error("listen %s:%d max:%d error\n", config->srv_addr, config->srv_port, config->cfg_max_clients);
		exit(1);
	}

	// 主循环
	for(;;)
	{
		readyfds=epoll_wait(config->srv_epfd,events,EPOLL_EVENT_SIZE,EPOLL_WAIT_TIMEOUT);

		if (readyfds < 0){
			chat_log_error("epoll_weit errno:%d %s\n", errno, strerror(errno));
		}

		// 请求时间处理
		memset(config->s_req_time, 0, SIZE_REQ_TIME);
		time(&now);
		now_time=localtime(&now);
		strftime( config->s_req_time, SIZE_REQ_TIME, "%Y-%m-%d %H:%M:%S", now_time);
		config->len_req_time = strlen(config->s_req_time);
		config->t_req_time = now;

		for(i=0;i<readyfds;++i)
		{
			chat_log_debug("event i=%d readyfds=%d\n", i, readyfds);

			if(events[i].data.fd==listenfd){// 新用户连接

				while(1){
					connfd = accept(listenfd, (struct sockaddr *)&clientaddr, &clilen);
					if(connfd<0){
						chat_log_debug("accept finish connfd<0 fd=%d errno:%d %s\n", connfd, errno, strerror(errno));
						break;
					}

					if (config->cur_clients >= config->cfg_max_clients){
						chat_log_debug("max clients, cur_client:%d max_client:%d\n", config->cur_clients, config->cfg_max_clients);
						close(connfd);
						continue;
					}

					setnonblocking(connfd);

					ev.data.fd=connfd;
					ev.events=EPOLLIN|EPOLLET;

					chat_log_debug("new clients fd=%d, num:%d %s\n", connfd, config->cur_clients, inet_ntoa(clientaddr.sin_addr));
					config->cur_clients++;

					if (config->stat_max_clients < config->cur_clients){
						config->stat_max_clients = config->cur_clients;
					}

					config->stat_last_fd = connfd;
					if (connfd > config->stat_max_fd)
						config->stat_max_fd = connfd;

					config->handler_new(connfd, &clientaddr, config);

					epoll_ctl(config->srv_epfd,EPOLL_CTL_ADD,connfd,&ev);
				}

			}else if(events[i].events & EPOLLIN){//如果是已经连接的用户，并且收到数据，那么进行读入。


				if ((sockfd = events[i].data.fd) < 0){
					chat_log_error("EPOLLIN fd < 0\n");
					continue;
				}

				session = config->fd_sessions[sockfd];
				if (session == NULL){
					chat_log_error("SESSION IS NULL fd=%d\n", sockfd);
					continue;
				}

				// 一次读完缓冲区
				old_buffer_data_size = session->read_buffer_data_size;
				buffersize = SESSION_READ_BUFFER_SIZE - session->read_buffer_data_size;
				while((nreadbytes = recv(sockfd, session->read_buffer+session->read_buffer_data_size, buffersize, 0)) > 0)
				{
					//chat_log_debug("<<recv fd:%d sb_size:%d old_b_size%d,b_size:%d, nreadbytes:%d errno:%d\n", 
					//			sockfd, SESSION_READ_BUFFER_SIZE, old_buffer_data_size, buffersize, nreadbytes, errno);

					session->read_buffer_data_size += nreadbytes;
					buffersize = SESSION_READ_BUFFER_SIZE - session->read_buffer_data_size;
					//chat_log_debug("<<recv after fd:%d sb_size:%d old_b_size%d,b_size:%d, nreadbytes:%d errno:%d\n", 
				//			sockfd, SESSION_READ_BUFFER_SIZE, session->read_buffer_data_size, buffersize, nreadbytes, errno);

					if (buffersize <= 0){// 缓冲区满，则断开连接
						chat_epoll_close(config, sockfd, CLOSE_BUFF_FULL);
						goto nextevent;
					}
				}

				// 有读到数据，则处理
				if (session->read_buffer_data_size > old_buffer_data_size){
					//chat_log_debug("handler_read start buffersize:%d fd:%d\n", session->read_buffer_data_size - old_buffer_data_size, sockfd);
					config->handler_read(session);
					old_buffer_data_size = session->read_buffer_data_size;
					//chat_log_debug("handler_read end buffersize:%d fd:%d\n", session->read_buffer_data_size - old_buffer_data_size, sockfd);
				}

				// 连接情况判断
				if (nreadbytes < 0 && errno == ECONNRESET){
					chat_epoll_close(config, sockfd, CLOSE_RSET);
				}else if (nreadbytes == 0){
					chat_epoll_close(config, sockfd, CLOSE_NORMAL);
				}

			}else if(events[i].events & EPOLLOUT){//如果是已经连接的用户，并且收到数据，那么进行读入。
				if ((sockfd = events[i].data.fd) < 0){
					chat_log_error("EPOLLOUT fd < 0\n");
					continue;
				}

				chat_log_debug("EPOLLOUT fd:%d\n", sockfd);

				session = config->fd_sessions[sockfd];
				if (session == NULL){
					chat_log_error("SESSION IS NULL EPOLLOUT\n");
					continue;
				}

				chat_send_item_t * item;

				tn = 0;
				while (session->send_head != NULL){
					item = session->send_head;

					n = item->buffer_size - item->buffer_data_size;
					writebytes = write_n(session->config, sockfd, item->buffer + item->buffer_data_size, n);
					tn+= writebytes;

					if (writebytes == n){// 写入成功
						session->send_head = item->next;
						if (session->send_head == NULL)
							session->send_last = NULL;

						// free
						if (item->is_free){
							if (item->buffer != NULL){
								free(item->buffer);
							}
						}

						free(item);
					}else{// 只有部分成功
						if (writebytes > 0){
							item->buffer_data_size += writebytes;
							break;
						}
					}
				}

				if (tn > 0){
					if (session->send_head == NULL){
						ev.data.fd=session->sockfd;
						ev.events = EPOLLIN | EPOLLET;
						epoll_ctl(session->config->srv_epfd,EPOLL_CTL_MOD,session->sockfd,&ev);
					}

					chat_log_debug(">>>EPOLLOUT write fd:%d tn:%d\n", sockfd, tn);
				}

			}else{// 其它异常情况，关闭
				sockfd = events[i].data.fd;
				chat_epoll_close(config, sockfd, CLOSE_EXCEPTION);
			}

			nextevent:;
		}

		// close disconnect session
		for(i=0; i < config->cfg_max_clients; i++){
			session = config->fd_sessions[i];

			if (session != NULL){
				if (session->flag_close){
					chat_epoll_close(config, i, CLOSE_WRITEFAIL);
				}else if (session->heart_time > 0 && (now - session->heart_time) > 30){
					chat_epoll_close(config, i, CLOSE_HEARTTIMEOUT);
				}else if (session->uid == NULL && (now - session->login_time) > cfg_maxconntime){
					chat_epoll_close(config, i, CLOSE_CONNTIMEOUT);
				}
			}
		}


	}
}

/**
 * 服务器端关闭连接
 */
void chat_epoll_close(chat_server_config_t *config, int sockfd, char *msg)
{
	int ret;

	ret = close(sockfd);

	if (ret == 0){
		if (config->cur_clients > 0)
			config->cur_clients--;

		config->handler_close(sockfd, config);
	}

	if (config->chat_phplog_fd != -1)	{
		char * buff_log = malloc(SIZE_LOG_BUFFER);
		int log_n = snprintf(buff_log, SIZE_LOG_BUFFER,
				"%s [close] fd:%d cur:%d  closeret:%d msg:%s\n",
				config->s_req_time, sockfd, config->cur_clients, ret, msg);
		chat_file_write(config->chat_phplog_fd, buff_log, log_n);
	}

	chat_log_debug("client close fd:%d cur:%d  closeret:%d msg:%s\n", sockfd, config->cur_clients, ret, msg);
}

/**
 * 设置非阻塞
 */
void setnonblocking(int sock)
{
	int opts;
	opts=fcntl(sock,F_GETFL);
	if(opts<0)
	{
		chat_log_error("fcntl(sock,F_GETFL) error\n");
		exit(1);
	}

	opts = opts|O_NONBLOCK;

	if(fcntl(sock,F_SETFL,opts)<0)
	{
		chat_log_error("fcntl(sock,SETFL,opts)");
		exit(1);
	}
}

/**
 * session发送消息
 */
ssize_t chat_send(chat_session_t *session, void *buffer, size_t n, int is_free)
{
	size_t      nleft;
	ssize_t     nwritten;
	static struct epoll_event ev;
	chat_send_item_t * item = NULL;
	int         close_flag;
	close_flag = 0;

	void *srcbuffer = buffer;
	nleft = n;

if (session->config->chat_phplog_fd != -1){
                char * buff_log = malloc(SIZE_LOG_BUFFER);
                int log_n = snprintf(buff_log, SIZE_LOG_BUFFER,
                "chat_send %s recv_suid=%s recv_snick=%s buffer_len=%zu\r\n",
                session->config->s_req_time, session->uid, session->nickname, n);
                chat_file_write(session->config->chat_phplog_fd, buff_log, log_n);
            }

	// 如果还有缓冲未发送，则直接加入后面
	if (session->send_last != NULL && session->flag_close == 0){
		chat_log_debug(">>write to buffer");

		// 注册EPOLLOUT事件
		ev.data.fd=session->sockfd;
		ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
		epoll_ctl(session->config->srv_epfd,EPOLL_CTL_MOD,session->sockfd,&ev);

		// 添加buff 到发送列表
		item = (chat_send_item_t *)malloc(sizeof(chat_send_item_t));
		item->buffer = (char *)srcbuffer;
		item->buffer_size = n;
		item->buffer_data_size = n- nleft;
		item->is_free = is_free;
		item->next = NULL;

		if (session->send_head == NULL){
			session->send_head = item;
		}
		if (session->send_last == NULL){
			session->send_last = item;
		}else{
			session->send_last->next = item;
			session->send_last = item;
		}

		return(n - nleft);
	}


	while (nleft > 0 && session->flag_close == 0) {
		if ((nwritten = write(session->sockfd, buffer, nleft)) < 0) {

			chat_log_debug(">>write fail nwriten=%d, fd=%d, errno=%d, error:%s\n", nwritten, session->sockfd, errno, strerror(errno));

			if (errno == EAGAIN || errno == EINTR){// 缓冲区满，继续写入

				// 注册EPOLLOUT事件
				ev.data.fd=session->sockfd;
				ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
				epoll_ctl(session->config->srv_epfd,EPOLL_CTL_MOD,session->sockfd,&ev);

				// 添加buff 到发送列表
				item = (chat_send_item_t *)malloc(sizeof(chat_send_item_t));
				item->buffer = (char *)srcbuffer;
				item->buffer_size = n;
				item->buffer_data_size = n- nleft;
				item->is_free = is_free;
				item->next = NULL;

				if (session->send_head == NULL){
					session->send_head = item;
				}
				if (session->send_last == NULL){
					session->send_last = item;
				}else{
					session->send_last->next = item;
					session->send_last = item;
				}
			}else if (errno == EPIPE){ // 连接断开执行关闭操作
				close_flag = 1;
			}

			break;
		}

		if (nwritten == 0)
			break;

		nleft -= nwritten;
		buffer   = (char *)buffer + nwritten;
	}


	if (item == NULL && is_free){
		if (srcbuffer != NULL)
			free(srcbuffer);

		if (nleft > 0){
			chat_log_debug(">>write fail fd=%d n:%d errno:%d %s\n", session->sockfd, n, errno, strerror(errno));
		}
	}

	// 标记断开，直接断开会有异常，后续包无法处理
	if (close_flag){
		session->flag_close = 1;
	}

	return(n - nleft);      /* return >= 0 */
}


static ssize_t write_n(chat_server_config_t *config, int fd, const void *buffer, size_t n)
{
	size_t      nleft;
	ssize_t     nwritten;
	int close_flag;
	close_flag = 0;

	nleft = n;
	while (nleft > 0) {
		if ((nwritten = write(fd, buffer, nleft)) < 0) {
			if (errno == EPIPE){
				close_flag = 1;
			}

			break;
		}

		if (nwritten == 0)
			break;

		nleft -= nwritten;
		buffer   = (char *)buffer + nwritten;
	}

	if (close_flag){
		chat_epoll_close(config, fd, CLOSE_BROKENPIPE);
		nleft = n;
	}

	return(n - nleft);      /* return >= 0 */
}
